package com.alibaba.nacos.core.remote.grpc;

import com.alibaba.nacos.api.common.Constants;
import com.alibaba.nacos.api.grpc.auto.BiRequestStreamGrpc;
import com.alibaba.nacos.api.grpc.auto.Payload;
import com.alibaba.nacos.api.remote.request.ConnectionSetupRequest;
import com.alibaba.nacos.api.remote.response.Response;
import com.alibaba.nacos.common.remote.ConnectionType;
import com.alibaba.nacos.common.remote.client.grpc.GrpcUtils;
import com.alibaba.nacos.core.remote.ConnectionManager;
import com.alibaba.nacos.core.remote.ConnectionMeta;
import com.alibaba.nacos.core.remote.RpcAckCallbackSynchronizer;
import io.grpc.netty.shaded.io.netty.channel.Channel;
import io.grpc.stub.StreamObserver;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.collections4.MapUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

import java.util.Map;

@Service
@Slf4j
public class GrpcBiStreamRequestAcceptor
        extends BiRequestStreamGrpc.BiRequestStreamImplBase {

    @Autowired
    private ConnectionManager connectionManager;

    @Override
    public StreamObserver<Payload> requestBiStream(StreamObserver<Payload> emitter) {
        return new StreamObserver<>() {
            final String connectionId = GrpcServerConstants.CONTEXT_KEY_CONN_ID.get();
            final int localPort = GrpcServerConstants.CONTEXT_KEY_CONN_LOCAL_PORT.get();
            final int remotePort = GrpcServerConstants.CONTEXT_KEY_CONN_REMOTE_PORT.get();
            String remoteIp = GrpcServerConstants.CONTEXT_KEY_CONN_REMOTE_IP.get();
            String clientIp = "";

            @Override
            public void onNext(Payload req) {
                clientIp = req.getMetadata().getClientIp();
                Object parseObj = GrpcUtils.parse(req);
                if (parseObj == null) {
                    log.error("[{}]Grpc request bi stream,req parse null ,body={},meta={}", connectionId, req.getBody().getValue().toStringUtf8(), req.getMetadata());
                    return;
                }

                if (parseObj instanceof ConnectionSetupRequest setupRequest) {
                    Map<String, String> labels = setupRequest.getLabels();
                    String appName = MapUtils.getString(labels, Constants.APPNAME, "-");
                    ConnectionMeta connectionMeta = new ConnectionMeta().setConnectionId(connectionId)
                            .setClientIp(req.getMetadata().getClientIp())
                            .setRemoteIp(remoteIp)
                            .setRemotePort(remotePort)
                            .setLocalPort(localPort)
                            .setConnectType(ConnectionType.GRPC.getType())
                            .setVersion(setupRequest.getClientVersion())
                            .setAppName(appName)
                            .setLabels(setupRequest.getLabels())
                            .setTenant(setupRequest.getTenant());

                    Channel channel = GrpcServerConstants.CONTEXT_KEY_CHANNEL.get();
                    GrpcConnection connection = new GrpcConnection(connectionMeta, emitter, channel);
                    connection.setAbilityTable(setupRequest.getAbilityTable());

                    boolean register = connectionManager.register(connectionId, connection);
                    if (!register) {
                        log.error("{}:Connection register fail, reason:{}", connectionId, "server is over limited");
                        connection.close();
                    } else {
                        // todo
                    }
                } else if (parseObj instanceof Response response) {
                    RpcAckCallbackSynchronizer.ackNotify(connectionId, response);
                    connectionManager.refreshActiveTime(connectionId);
                }
            }

            @Override
            public void onError(Throwable t) {
                log.info("onError {}", t.getMessage());
            }

            @Override
            public void onCompleted() {
                log.info("onCompleted");
            }
        };
    }
}
